package com.bw.search;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;

public class EcommerceRecommendationSystem {

    // 商品点击事件类
    public static class ItemViewEvent {
        private Long userId;
        private Long itemId;
        private Integer categoryId;
        private String behavior;
        private Long timestamp;

        public ItemViewEvent() {}

        public ItemViewEvent(Long userId, Long itemId, Integer categoryId, String behavior, Long timestamp) {
            this.userId = userId;
            this.itemId = itemId;
            this.categoryId = categoryId;
            this.behavior = behavior;
            this.timestamp = timestamp;
        }

        // getters and setters
        public Long getUserId() { return userId; }
        public void setUserId(Long userId) { this.userId = userId; }
        public Long getItemId() { return itemId; }
        public void setItemId(Long itemId) { this.itemId = itemId; }
        public Integer getCategoryId() { return categoryId; }
        public void setCategoryId(Integer categoryId) { this.categoryId = categoryId; }
        public String getBehavior() { return behavior; }
        public void setBehavior(String behavior) { this.behavior = behavior; }
        public Long getTimestamp() { return timestamp; }
        public void setTimestamp(Long timestamp) { this.timestamp = timestamp; }

        @Override
        public String toString() {
            return "ItemViewEvent{" +
                    "userId=" + userId +
                    ", itemId=" + itemId +
                    ", categoryId=" + categoryId +
                    ", behavior='" + behavior + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }

    // 商品统计信息类
    public static class ItemViewCount {
        private Long itemId;
        private Long windowEnd;
        private Long count;

        public ItemViewCount() {}

        public ItemViewCount(Long itemId, Long windowEnd, Long count) {
            this.itemId = itemId;
            this.windowEnd = windowEnd;
            this.count = count;
        }

        // getters and setters
        public Long getItemId() { return itemId; }
        public void setItemId(Long itemId) { this.itemId = itemId; }
        public Long getWindowEnd() { return windowEnd; }
        public void setWindowEnd(Long windowEnd) { this.windowEnd = windowEnd; }
        public Long getCount() { return count; }
        public void setCount(Long count) { this.count = count; }

        @Override
        public String toString() {
            return "ItemViewCount{" +
                    "itemId=" + itemId +
                    ", windowEnd=" + windowEnd +
                    ", count=" + count +
                    '}';
        }
    }

    // 商品推荐结果类
    public static class RecommendationResult {
        private Long userId;
        private Long itemId;
        private double score;
        private String reason;
        private Long timestamp;

        public RecommendationResult() {}

        public RecommendationResult(Long userId, Long itemId, double score, String reason, Long timestamp) {
            this.userId = userId;
            this.itemId = itemId;
            this.score = score;
            this.reason = reason;
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return "推荐结果: 用户ID=" + userId +
                    ", 商品ID=" + itemId +
                    ", 推荐分数=" + score +
                    ", 推荐理由='" + reason + '\'' +
                    ", 时间戳=" + timestamp;
        }

        // getters and setters
        public Long getUserId() { return userId; }
        public void setUserId(Long userId) { this.userId = userId; }
        public Long getItemId() { return itemId; }
        public void setItemId(Long itemId) { this.itemId = itemId; }
        public double getScore() { return score; }
        public void setScore(double score) { this.score = score; }
        public String getReason() { return reason; }
        public void setReason(String reason) { this.reason = reason; }
        public Long getTimestamp() { return timestamp; }
        public void setTimestamp(Long timestamp) { this.timestamp = timestamp; }
    }

    // 模拟数据生成器
    public static class SimulatedDataGenerator implements SourceFunction<ItemViewEvent> {
        private static final long serialVersionUID = 1L;
        private volatile boolean isRunning = true;
        private final Random random = new Random();
        private final int numUsers = 1000; // 模拟1000个用户
        private final int numItems = 10000; // 模拟10000个商品
        private final int numCategories = 20; // 模拟20个商品类别
        private final String[] behaviors = {"pv", "buy", "cart", "fav"}; // 用户行为类型
        private long currentTimestamp = System.currentTimeMillis() / 1000; // 秒级时间戳

        @Override
        public void run(SourceContext<ItemViewEvent> ctx) throws Exception {
            while (isRunning) {
                // 生成随机用户行为
                long userId = ThreadLocalRandom.current().nextLong(numUsers) + 1;
                long itemId = ThreadLocalRandom.current().nextLong(numUsers) + 1;;
                int categoryId = (int) ((itemId % numCategories) + 1);
                String behavior = behaviors[random.nextInt(behaviors.length)];
                
                // 时间戳递增，模拟顺序事件
                currentTimestamp += random.nextInt(5);
                
                // 生成有偏好的行为数据（用户更倾向于浏览某些类别）
                if (random.nextDouble() < 0.7) {
                    // 用户的偏好类别（每个用户有2个偏好类别）
                    int preferredCategory1 = (int) (userId % numCategories) + 1;
                    int preferredCategory2 = (int) ((userId / 2) % numCategories) + 1;
                    
                    // 70%的概率浏览偏好类别中的商品
                    if (random.nextBoolean()) {
                        // 修正：使用ThreadLocalRandom生成指定范围的随机数
                        itemId = (preferredCategory1 - 1) * (numItems / numCategories) +
                                ThreadLocalRandom.current().nextLong(numItems / numCategories) + 1;
                        categoryId = preferredCategory1;
                    } else {
                        itemId = (preferredCategory2 - 1) * (numItems / numCategories) +
                                ThreadLocalRandom.current().nextLong(numItems / numCategories) + 1;
                        categoryId = preferredCategory2;
                    }
                }
                
                // 生成点击行为的概率更高
                if (random.nextDouble() < 0.6) {
                    behavior = "pv";
                }
                
                // 输出事件
                ctx.collect(new ItemViewEvent(userId, itemId, categoryId, behavior, currentTimestamp));
                
                // 控制生成速率
                Thread.sleep(1000);
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }

    // 热门商品统计
    public static class HotItemsAnalyzer extends RichFlatMapFunction<ItemViewEvent, ItemViewCount> {
        private MapState<Long, Long> itemCountState; // 商品ID -> 点击量
        private long windowSize = 3600; // 1小时窗口
        private long slideSize = 600; // 10分钟滑动步长

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            itemCountState = getRuntimeContext().getMapState(
                new MapStateDescriptor<>("itemCountState", Long.class, Long.class)
            );
        }

        @Override
        public void flatMap(ItemViewEvent event, Collector<ItemViewCount> out) throws Exception {
            // 只处理点击行为
            if ("pv".equals(event.getBehavior())) {
                // 更新商品点击量
                Long count = itemCountState.get(event.getItemId());
                if (count == null) {
                    count = 0L;
                }
                count++;
                itemCountState.put(event.getItemId(), count);
                
                // 模拟窗口输出（实际生产中应使用真正的窗口操作）
                if (event.getTimestamp() % slideSize == 0) {
                    long windowEnd = event.getTimestamp() / slideSize * slideSize + slideSize;
                    
                    // 输出当前窗口的热门商品
                    for (Map.Entry<Long, Long> entry : itemCountState.entries()) {
                        out.collect(new ItemViewCount(entry.getKey(), windowEnd, entry.getValue()));
                    }
                    
                    // 清除过期状态（模拟窗口滑动）
                    if (itemCountState.keys().iterator().hasNext()) {
                        itemCountState.clear();
                    }
                }
            }
        }
    }

    // 基于用户行为的推荐引擎
    public static class RecommendationEngine extends RichFlatMapFunction<ItemViewEvent, RecommendationResult> {
        private MapState<Long, List<Long>> userBehaviorState; // 用户ID -> 浏览过的商品列表
        private MapState<Long, Map<Long, Integer>> itemCooccurrenceState; // 商品ID -> {关联商品ID -> 共现次数}
        private MapState<Long, Integer> categoryPopularityState; // 类别ID -> 热度值

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            userBehaviorState = getRuntimeContext().getMapState(
                    new MapStateDescriptor<>(
                            "userBehaviorState",
                            TypeInformation.of(Long.class),  // 键类型
                            TypeInformation.of(new TypeHint<List<Long>>() {})  // 值类型
                    )
            );

            itemCooccurrenceState = getRuntimeContext().getMapState(
                    new MapStateDescriptor<>(
                            "itemCooccurrenceState",
                            TypeInformation.of(Long.class),  // 键类型
                            TypeInformation.of(new TypeHint<Map<Long, Integer>>() {})  // 值类型
                    )
            );
            categoryPopularityState = getRuntimeContext().getMapState(
                new MapStateDescriptor<>("categoryPopularityState", Long.class, Integer.class)
            );
        }

        @Override
        public void flatMap(ItemViewEvent event, Collector<RecommendationResult> out) throws Exception {
            Long userId = event.getUserId();
            Long itemId = event.getItemId();
            Integer categoryId = event.getCategoryId();
            String behavior = event.getBehavior();
            
            // 更新类别热度
            Integer categoryCount = categoryPopularityState.get(Long.valueOf(categoryId));
            if (categoryCount == null) {
                categoryCount = 0;
            }
            // 不同行为赋予不同权重
            int behaviorWeight = getBehaviorWeight(behavior);
            categoryPopularityState.put(Long.valueOf(categoryId), categoryCount + behaviorWeight);
            
            // 更新用户行为历史
            List<Long> userHistory = userBehaviorState.get(userId);
            if (userHistory == null) {
                userHistory = new ArrayList<>();
            }
            
            // 只记录浏览、收藏和购买行为
            if ("pv".equals(behavior) || "fav".equals(behavior) || "buy".equals(behavior)) {
                // 避免重复记录
                if (!userHistory.contains(itemId)) {
                    userHistory.add(itemId);
                    
                    // 限制历史记录长度
                    if (userHistory.size() > 20) {
                        userHistory = userHistory.subList(userHistory.size() - 20, userHistory.size());
                    }
                    
                    userBehaviorState.put(userId, userHistory);
                    
                    // 更新商品共现矩阵
                    updateItemCooccurrence(itemId, userHistory);
                    
                    // 生成推荐
                    generateRecommendations(userId, itemId, categoryId, event.getTimestamp(), out);
                }
            }
        }

        private int getBehaviorWeight(String behavior) {
            switch (behavior) {
                case "buy": return 5;
                case "fav": return 3;
                case "cart": return 2;
                case "pv": return 1;
                default: return 1;
            }
        }

        /**
         * A  B  C  D
         *
         * userHistory:A,B
         *
         * C
         *
         * @param currentItem
         * @param userHistory
         * @throws Exception
         */
        private void updateItemCooccurrence(Long currentItem, List<Long> userHistory) throws Exception {
            // 更新当前商品与用户历史中商品的共现关系
            for (Long viewedItem : userHistory) {
                if (!viewedItem.equals(currentItem)) {
                    // 更新viewedItem -> currentItem的共现次数
                    Map<Long, Integer> cooccurrenceMap = itemCooccurrenceState.get(viewedItem);
                    if (cooccurrenceMap == null) {
                        cooccurrenceMap = new HashMap<>();
                    }

                    cooccurrenceMap.put(currentItem, cooccurrenceMap.getOrDefault(currentItem, 0) + 1);
                    itemCooccurrenceState.put(viewedItem, cooccurrenceMap); // A:{"B":1}
                    
                    // 更新currentItem -> viewedItem的共现次数
                    cooccurrenceMap = itemCooccurrenceState.get(currentItem);
                    if (cooccurrenceMap == null) {
                        cooccurrenceMap = new HashMap<>();
                    }

                    cooccurrenceMap.put(viewedItem, cooccurrenceMap.getOrDefault(viewedItem, 0) + 1);
                    itemCooccurrenceState.put(currentItem, cooccurrenceMap);// B:{"A":1}
                }
            }

        }

        private void generateRecommendations(Long userId, Long currentItem, Integer categoryId, long timestamp, Collector<RecommendationResult> out) throws Exception {
            List<Long> userHistory = userBehaviorState.get(userId);
            if (userHistory == null || userHistory.isEmpty()) {
                // 用户没有历史行为，推荐热门商品
                recommendPopularItems(userId, categoryId, timestamp, out);
                return;
            }
            
            // 基于用户历史行为的协同过滤推荐
            Map<Long, Double> recommendationScores = new HashMap<>();
            
            // 为每个历史商品查找相似商品
            for (Long item : userHistory) {
                Map<Long, Integer> cooccurrenceMap = itemCooccurrenceState.get(item);
                if (cooccurrenceMap != null) {
                    for (Map.Entry<Long, Integer> entry : cooccurrenceMap.entrySet()) {
                        Long relatedItem = entry.getKey();
                        int cooccurrenceCount = entry.getValue();
                        
                        // 避免推荐用户已经浏览过的商品
                        if (!userHistory.contains(relatedItem)) {
                            // 计算推荐分数，考虑共现次数和用户历史商品的权重
                            double score = cooccurrenceCount * getBehaviorWeightForItem(item, userHistory);
                            recommendationScores.put(relatedItem, recommendationScores.getOrDefault(relatedItem, 0.0) + score);
                        }
                    }
                }
            }
            
            // 按分数排序并输出推荐结果
            List<Map.Entry<Long, Double>> sortedRecommendations = recommendationScores.entrySet()
                .stream()
                .sorted(Map.Entry.<Long, Double>comparingByValue().reversed())
                .limit(5)
                .collect(Collectors.toList());
            
            for (Map.Entry<Long, Double> entry : sortedRecommendations) {
                out.collect(new RecommendationResult(
                    userId, 
                    entry.getKey(), 
                    entry.getValue(), 
                    "基于您浏览的商品推荐", 
                    timestamp
                ));
            }
            
            // 如果协同过滤推荐不足5个，补充热门商品
            if (sortedRecommendations.size() < 5) {
                recommendPopularItems(userId, categoryId, timestamp, out, 5 - sortedRecommendations.size());
            }
        }

        private double getBehaviorWeightForItem(Long item, List<Long> userHistory) {
            // 假设用户历史中的最后一个商品权重最高，依次递减
            int index = userHistory.indexOf(item);
            return 1.0 - (double) index / userHistory.size();
        }

        private void recommendPopularItems(Long userId, Integer categoryId, long timestamp, Collector<RecommendationResult> out) throws Exception {
            recommendPopularItems(userId, categoryId, timestamp, out, 5);
        }

        private void recommendPopularItems(Long userId, Integer categoryId, long timestamp, Collector<RecommendationResult> out, int count) throws Exception {
            // 获取热门类别
//            List<Map.Entry<Long, Integer>> popularCategories = new ArrayList<>(categoryPopularityState.entries());
            List<Map.Entry<Long, Integer>> popularCategories = new ArrayList<>();
            for (Map.Entry<Long, Integer> entry : categoryPopularityState.entries()) {
                popularCategories.add(entry);
            }
            popularCategories.sort(Map.Entry.<Long, Integer>comparingByValue().reversed());
            
            // 基于热门类别生成推荐
            for (Map.Entry<Long, Integer> category : popularCategories) {
                if (count <= 0) break;
                
                // 模拟生成该类别下的热门商品
                long fakeItemId = (category.getKey() - 1) * 1000 + ThreadLocalRandom.current().nextLong(1000) + 1;
                
                // 避免推荐用户已经浏览过的商品
                List<Long> userHistory = userBehaviorState.get(userId);
                if (userHistory == null || !userHistory.contains(fakeItemId)) {
                    out.collect(new RecommendationResult(
                        userId, 
                        fakeItemId, 
                        category.getValue() * 0.1, 
                        "热门商品推荐", 
                        timestamp
                    ));
                    count--;
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        // 设置事件时间语义,生成水位线间隔
        env.getConfig().setAutoWatermarkInterval(1000L);

        // 生成模拟数据流
        DataStream<ItemViewEvent> inputStream = env.addSource(new SimulatedDataGenerator())
            .assignTimestampsAndWatermarks(WatermarkStrategy.<ItemViewEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<ItemViewEvent>() {
                    @Override
                    public long extractTimestamp(ItemViewEvent event, long l) {
                        return event.getTimestamp() * 1000; // 转换为毫秒
                    }
                }));

//        inputStream.print();

//        // 1. 热门商品统计
        DataStream<ItemViewCount> hotItemsStream = inputStream
                .keyBy(ItemViewEvent::getItemId)
                .flatMap(new HotItemsAnalyzer());
//        hotItemsStream.print();
//
//        // 2. 商品推荐引擎
        DataStream<RecommendationResult> recommendationStream = inputStream
                .keyBy(ItemViewEvent::getUserId)
                .flatMap(new RecommendationEngine());
//        recommendationStream.print();
//
//        // 打印热门商品统计结果
//        hotItemsStream
//            .keyBy(ItemViewCount::getWindowEnd)
//            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//            .process(new TopNHotItems(3)) // 计算Top3热门商品
//            .print("热门商品统计");
//
//        // 打印推荐结果
        recommendationStream.print("个性化推荐");

        // 执行作业
        env.execute("电商搜索推荐系统");
    }

    // 计算TopN热门商品
    public static class TopNHotItems extends ProcessWindowFunction<ItemViewCount, String, Long, TimeWindow> {
        private final int topSize;

        public TopNHotItems(int topSize) {
            this.topSize = topSize;
        }

        @Override
        public void process(Long windowEnd, Context context, Iterable<ItemViewCount> elements, Collector<String> out) throws Exception {
            // 收集所有商品及其点击量
            List<ItemViewCount> itemCounts = new ArrayList<>();
            elements.forEach(itemCounts::add);
            
            // 按点击量排序
            itemCounts.sort((o1, o2) -> o2.getCount().compareTo(o1.getCount()));
            
            // 构建输出结果
            StringBuilder result = new StringBuilder();
            result.append("====================================\n");
            result.append("窗口结束时间: ").append(new Date(windowEnd)).append("\n");
            
            // 取TopN
            for (int i = 0; i < Math.min(topSize, itemCounts.size()); i++) {
                ItemViewCount item = itemCounts.get(i);
                result.append("Top ").append(i + 1).append(": ")
                    .append("商品ID=").append(item.getItemId())
                    .append(" 点击量=").append(item.getCount())
                    .append("\n");
            }
            result.append("====================================\n\n");
            
            out.collect(result.toString());
        }
    }
}    